package com.example.kafka.component;

import com.example.kafka.listener.IOTListener;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.CommonErrorHandler;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Component;

import java.util.Map;

@Slf4j
@Component
public class StartIOTListener implements ApplicationRunner {

    private static final String GROUP_ID = "work-assistant";

    private ConsumerFactory<String, String> factory;

    private ObjectMapper objectMapper;

    @Autowired
    public StartIOTListener(ConsumerFactory<String, String> factory, ObjectMapper objectMapper){
        this.factory = factory;
        this.objectMapper = objectMapper;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        Map<String, Object> properties = factory.getConfigurationProperties();
        if (properties.isEmpty()){
            log.info("未检测到kafka相关配置 无法启动");
            return;
        }
        IOTListener iotListener = new IOTListener(objectMapper, GROUP_ID);
        ConcurrentMessageListenerContainer<String, String> container =
                new ConcurrentMessageListenerContainer<>(factory, iotListener.getContainerProperties(iotListener.getGroupId(), null, 0));
        //禁止KafkaListener自启动 未配置kafka的环境不启用
        container.setAutoStartup(false);
        //设置并发消费的线程数量 为1个
        container.setConcurrency(1);
        //全局的异常消息消费处理器
        container.setCommonErrorHandler(new CommonErrorHandler() {
            @Override
            public void handleRecord(@NonNull Exception e, @NonNull ConsumerRecord<?, ?> consumerRecord,
                                     @NonNull Consumer<?, ?> consumer, @NonNull MessageListenerContainer container) {
                log.error("kafka消费异常", e);
                log.info("异常消息:{}, 异常消息主题: {} ,消息offset: {} ,消息分区: {}",
                        consumerRecord.value(),
                        consumerRecord.topic(),
                        consumerRecord.offset(),
                        consumerRecord.partition());
                /**
                 * 由于开启了手动提交偏移量
                 * 所以异常处理需要调用当前方法
                 */
                consumer.commitAsync();
            }
        });
        //启动
        container.start();
    }
}
